package com.group1.realtime.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.group1.realtime.app.BaseAppV2;
import com.group1.realtime.bean.OrderWithOutSourceBean;
import com.group1.realtime.bean.OrderWithSourceBean;
import com.group1.realtime.bean.PageWithSourceBean;
import com.group1.realtime.function.DimAsyncFunction;
import com.group1.realtime.util.AtguiguUtil;
import com.group1.realtime.util.FlinkSinkUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.JoinFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.TimeUnit;

import static com.group1.realtime.common.Constant.TOPIC_DWD_TRAFFIC_PAGE;
import static com.group1.realtime.common.Constant.TOPIC_ODS_DB;

public class DwsTradeOrderSourceInfo extends BaseAppV2 {
    public static void main(String[] args) {
        new DwsTradeOrderSourceInfo().init(
                2003,
                2,
                "DwsTradeOrderSourceInfo",
                TOPIC_ODS_DB,
                TOPIC_DWD_TRAFFIC_PAGE
        );
    }

    @Override
    protected void handle(StreamExecutionEnvironment env, HashMap<String, DataStreamSource<String>> topicAndStreamMap) {
        DataStream<OrderWithSourceBean> beanStream = parseAndUnionOne(topicAndStreamMap);
        SingleOutputStreamOperator<OrderWithSourceBean> aggStream = windowAndAgg(beanStream);
        aggStream.print();
        SingleOutputStreamOperator<OrderWithSourceBean> withDimStream = addDim(aggStream);
//        withDimStream.print();
//        writeToClickHouse(aggStream);
    }

    private void writeToClickHouse(SingleOutputStreamOperator<OrderWithSourceBean> stream) {
        stream.addSink(FlinkSinkUtil.getClickHouseSink("dws_trade_order_source_window", OrderWithSourceBean.class));
    }

    private SingleOutputStreamOperator<OrderWithSourceBean> addDim(SingleOutputStreamOperator<OrderWithSourceBean> beanStreamWithoutDim) {
        SingleOutputStreamOperator<OrderWithSourceBean> sourceSiteStream = AsyncDataStream.unorderedWait(
                beanStreamWithoutDim,
                new DimAsyncFunction<OrderWithSourceBean>() {
                    @Override
                    public String getTable() {
                        return "dim_source";
                    }

                    @Override
                    public String getId(OrderWithSourceBean input) {
                        return input.getSource();
                    }

                    @Override
                    public void addDim(OrderWithSourceBean input, JSONObject dim) {
                        input.setSourceName(dim.getString("SOURCE_SITE"));
                    }
                },
                60,
                TimeUnit.SECONDS
        );
        return AsyncDataStream.unorderedWait(
                sourceSiteStream,
                new DimAsyncFunction<OrderWithSourceBean>() {
                    @Override
                    public String getTable() {
                        return "dim_province";
                    }

                    @Override
                    public String getId(OrderWithSourceBean input) {
                        return input.getProvinceId();
                    }

                    @Override
                    public void addDim(OrderWithSourceBean input, JSONObject dim) {
                        input.setProvinceName(dim.getString("NAME"));
                    }
                },
                60,
                TimeUnit.SECONDS
        );
    }


    private SingleOutputStreamOperator<OrderWithSourceBean> windowAndAgg(DataStream<OrderWithSourceBean> beanStream) {
        return beanStream
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy
                                .<OrderWithSourceBean>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                                .withTimestampAssigner((bean, ts) -> bean.getTs())
                )
//                .keyBy(OrderWithSourceBean::getOrderId)
                .keyBy(OrderWithSourceBean::getSource)
                .window(TumblingEventTimeWindows.of(Time.seconds(5)))
                .reduce(
                        new ReduceFunction<OrderWithSourceBean>() {
                            @Override
                            public OrderWithSourceBean reduce(OrderWithSourceBean value1, OrderWithSourceBean value2) throws Exception {
                                value1.setFinalAmount(value1.getFinalAmount() + value2.getFinalAmount());
                                value1.setOrderCt(value1.getOrderCt() + value2.getOrderCt());

                                return value1;
                            }
                        },
                        new ProcessWindowFunction<OrderWithSourceBean, OrderWithSourceBean, String, TimeWindow>() {
                            @Override
                            public void process(String key, Context ctx,
                                                Iterable<OrderWithSourceBean> elements, Collector<OrderWithSourceBean> out) throws Exception {
                                OrderWithSourceBean bean = elements.iterator().next();
                                bean.setStt(AtguiguUtil.toDateTime(ctx.window().getStart()));
                                bean.setEdt(AtguiguUtil.toDateTime(ctx.window().getEnd()));

                                bean.setTs(ctx.currentProcessingTime());  // 把ts更新为聚合时间

                                out.collect(bean);
                            }
                        }
                );


    }

    private DataStream<OrderWithSourceBean> parseAndUnionOne(Map<String, DataStreamSource<String>> topicAndStreamMap) {
        return getOrderWithSourceStream(topicAndStreamMap);

    }

    DataStream<OrderWithSourceBean> getOrderWithSourceStream(Map<String, DataStreamSource<String>> topicAndStreamMap) {
        SingleOutputStreamOperator<OrderWithOutSourceBean> orderStream = topicAndStreamMap
                .get(TOPIC_ODS_DB)
                .filter(db -> "order_info".equals(JSON.parseObject(db).getString("table")))
                .map(new MapFunction<String, OrderWithOutSourceBean>() {
                    @Override
                    public OrderWithOutSourceBean map(String oi) throws Exception {
                        JSONObject obj = JSON.parseObject(oi);
                        String orderId = obj.getJSONObject("data").getString("id");
                        String usrId = obj.getJSONObject("data").getString("user_id");
                        Long finalAmount = obj.getJSONObject("data").getLong("final_amount");
                        String provinceId = obj.getJSONObject("data").getString("province_id");
                        Long ts = obj.getLong("ts");

                        return new OrderWithOutSourceBean(orderId, usrId, finalAmount, 1L, provinceId, ts * 1000);
                    }
                })
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy
                                .<OrderWithOutSourceBean>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                                .withTimestampAssigner((bean, ts) -> bean.getTs())
                );


        SingleOutputStreamOperator<PageWithSourceBean> pageStream = topicAndStreamMap
                .get(TOPIC_DWD_TRAFFIC_PAGE)
                .filter(page -> "order".equals(JSON.parseObject(page).getJSONObject("page").getString("page_id"))
                        && "order_id".equals(JSON.parseObject(page).getJSONObject("page").getString("item_type")))
                .map(new MapFunction<String, PageWithSourceBean>() {
                    @Override
                    public PageWithSourceBean map(String orderPage) throws Exception {
                        JSONObject common = JSON.parseObject(orderPage).getJSONObject("common");
                        JSONObject page = JSON.parseObject(orderPage).getJSONObject("page");
                        Long ts = JSON.parseObject(orderPage).getLong("ts");

                        String source = common.getString("sc");
                        String userId = common.getString("uid");
                        String orderId = page.getString("item");

                        return new PageWithSourceBean(orderId, userId, source, ts);
                    }
                })
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy
                                .<PageWithSourceBean>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                                .withTimestampAssigner((bean, ts) -> bean.getTs())
                );


        return orderStream
                .join(pageStream)
                .where(OrderWithOutSourceBean::getOrderId)
                .equalTo(PageWithSourceBean::getOrderId)
                .window(TumblingEventTimeWindows.of(Time.seconds(5)))
                .apply(new JoinFunction<OrderWithOutSourceBean, PageWithSourceBean, OrderWithSourceBean>() {
                    @Override
                    public OrderWithSourceBean join(OrderWithOutSourceBean bean1, PageWithSourceBean bean2) throws Exception {
                        return new OrderWithSourceBean("", "", bean1.getOrderId(), bean1.getUserId(), bean1.getFinalAmount(), bean1.getOrderCt(), bean1.getProvinceId(),"", bean2.getSource(),"", bean1.getTs());
                    }
                });
    }


}
